package com.lry.it;

import com.lry.block.BlockEntry;
import com.lry.util.*;

import java.util.Comparator;
import java.util.Map;
import java.util.NoSuchElementException;

import static com.google.common.base.Preconditions.*;
import static com.lry.util.SizeOf.SIZE_OF_INT;
import static java.util.Objects.requireNonNull;

public class BlockIterator implements SeekingIterator<Slice,Slice>{


    private final SliceInput data;
    private final Slice restartPositions;
    private final int restartCount;
    private final Comparator<Slice> comparator;
    private BlockEntry nextEntry;

    public BlockIterator(Slice data, Slice restartPositions, Comparator<Slice> comparator)
    {
        requireNonNull(data, "data is null");
        requireNonNull(restartPositions, "restartPositions is null");
        checkArgument(restartPositions.length() % SIZE_OF_INT == 0, "restartPositions.readableBytes() must be a multiple of %s", SIZE_OF_INT);
        requireNonNull(comparator, "comparator is null");

        this.data = data.input();
        this.restartPositions = restartPositions.slice();
        restartCount = this.restartPositions.length() / SIZE_OF_INT;

        this.comparator = comparator;

        seekToFirst();
    }

    @Override
    public void seekToFirst() {
        if (restartCount > 0) {
            seekToRestartPosition(0);
        }
    }

    private void seekToRestartPosition(int restartPosition) {
        checkPositionIndex(restartPosition, restartCount, "restartPosition");

        int offset = restartPositions.getInt(restartPosition * SIZE_OF_INT);
        data.setPosition(offset);
        nextEntry = null;
        nextEntry = readEntry(data, null);
    }

    private BlockEntry readEntry(SliceInput data, BlockEntry previousEntry) {
        requireNonNull(data, "data is null");

        int sharedKeyLength = VariableLengthQuantity.readVariableLengthInt(data);
        int nonSharedKeyLength = VariableLengthQuantity.readVariableLengthInt(data);
        int valueLength = VariableLengthQuantity.readVariableLengthInt(data);

        // read key
        final Slice key;
        if (sharedKeyLength > 0) {
            key = Slices.allocate(sharedKeyLength + nonSharedKeyLength);
            SliceOutput sliceOutput = key.output();
            checkState(previousEntry != null, "Entry has a shared key but no previous entry was provided");
            sliceOutput.writeBytes(previousEntry.getKey(), 0, sharedKeyLength);
            sliceOutput.writeBytes(data, nonSharedKeyLength);
        }
        else {
            key = data.readSlice(nonSharedKeyLength);
        }
        // read value
        Slice value = data.readSlice(valueLength);

        return new BlockEntry(key, value);
    }

    @Override
    public void seek(Slice targetKey) {
        if (restartCount == 0) {
            return;
        }

        int left = 0;
        int right = restartCount - 1;

        // binary search restart positions to find the restart position immediately before the targetKey
        while (left < right) {
            int mid = (left + right + 1) / 2;

            seekToRestartPosition(mid);

            if (comparator.compare(nextEntry.getKey(), targetKey) < 0) {
                // key at mid is smaller than targetKey.  Therefore all restart
                // blocks before mid are uninteresting.
                left = mid;
            }
            else {
                // key at mid is greater than or equal to targetKey.  Therefore
                // all restart blocks at or after mid are uninteresting.
                right = mid - 1;
            }
        }

        // linear search (within restart block) for first key greater than or equal to targetKey
        for (seekToRestartPosition(left); nextEntry != null; next()) {
            if (comparator.compare(peek().getKey(), targetKey) >= 0) {
                break;
            }
        }
    }

    @Override
    public Map.Entry<Slice, Slice> peek() {
        if (!hasNext()) {
            throw new NoSuchElementException();
        }
        return nextEntry;
    }

    @Override
    public boolean hasNext() {
        return nextEntry != null;
    }

    @Override
    public Map.Entry<Slice, Slice> next() {
        if (!hasNext()) {
            throw new NoSuchElementException();
        }

        BlockEntry entry = nextEntry;

        if (!data.isReadable()) {
            nextEntry = null;
        }
        else {
            // read entry at current data position
            nextEntry = readEntry(data, nextEntry);
        }

        //返回上一个
        return entry;

    }

    @Override
    public void remove() {
        throw new UnsupportedOperationException();
    }
}
